package com.zetyun.tiger.cdc.oracle;

import com.fasterxml.jackson.annotation.JsonProperty;
import com.ververica.cdc.connectors.oracle.table.StartupOptions;
import com.zetyun.rt.meta.jsonschema.annotation.JsonSchemaTitle;
import lombok.Getter;
import lombok.Setter;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonPropertyDescription;

import java.util.Map;
import java.util.Properties;
import java.util.Set;

@Getter
@Setter
@ToString
@Slf4j
public class OracleCdcParameter {

    @JsonSchemaTitle("主机名/IP")
    @JsonProperty(defaultValue = "rdx2")
    private String hostName;


    @JsonSchemaTitle("端口/PORT")
    @JsonProperty(defaultValue = "1521")
    private Integer port;

    @JsonSchemaTitle("数据库/DATABASE")
    @JsonProperty(defaultValue = "orcl")
    private String database;

    @JsonSchemaTitle("schema列表")
    @JsonProperty(required = true)
    private Set<String> schemaList;

    @JsonSchemaTitle("table列表")
    @JsonProperty(required = true)
    private Set<String> tableList;

    @JsonSchemaTitle("用户名")
    @JsonProperty(defaultValue = "flinkuser")
    private String userName;

    @JsonSchemaTitle("密码")
    @JsonProperty(defaultValue = "flinkpw")
    private String passWord ;

    @JsonSchemaTitle("初始化读取?")
    @JsonPropertyDescription("启动模式？INITIAL:LATEST_OFFSET")
    private Boolean isInitial = false;

    @JsonSchemaTitle("oracle11g?")
    @JsonPropertyDescription("是否适配oracle11g")
    private Boolean isOracle11g = false;

    @JsonSchemaTitle("是否启用在线挖掘模式")
    @JsonPropertyDescription("可以使用在线挖掘的模式，不写入数据字典到 redo log 中，但是这样无法处理 DDL 语句。 生产环境默认策略读取 log 较慢，且默认策略会写入数据字典信息到 redo log 中导致日志量增加较多")
    private Boolean enableOnlineMiningMode = true;

    @JsonSchemaTitle("debezium配置")
    private Map<String, String> debeziumProp;

    public Properties toProp() {
        Properties prop = new Properties();
        if (debeziumProp != null && !debeziumProp.isEmpty()) {
            prop.putAll(debeziumProp);
        }
        if (isOracle11g) {
            prop.setProperty("database.tablename.case.insensitive", "false");
        }
        if (enableOnlineMiningMode) {
            prop.setProperty("log.mining.strategy", "online_catalog");
            prop.setProperty("log.mining.continuous.mine", "true");
        }
        return prop;
    }

    public StartupOptions getStartupOptions() {
        if (isInitial) {
            return StartupOptions.initial();
        } else {
            return StartupOptions.latest();
        }
    }

}
